/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.mapreduce.benchmark;

import static java.lang.Math.min;

import java.util.Arrays;
import java.util.Map;
import java.util.Properties;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.sss.client.SimpleJob;
import org.sss.client.SssClient;
import org.sss.mapreduce.Configuration;
import org.sss.mapreduce.GroupID;
import org.sss.mapreduce.Mapper;
import org.sss.mapreduce.Output;
import org.sss.mapreduce.Partitioner;
import org.sss.mapreduce.Reducer;
import org.sss.mapreduce.datatype.PackableBytes;
import org.sss.mapreduce.datatype.PackableString;

public class IoCompositeRun {
  private static Logger logger = LoggerFactory.getLogger(IoCompositeRun.class);
// for debug -->
  private static int mapperCalled = 0;
  private static int reducerCalled = 0;
  private static synchronized void mapperCalled() {
    mapperCalled++;
  }
  @SuppressWarnings("unused")
  private static synchronized void reducerCalled() {
    reducerCalled++;
  }
// for debug <--

  private static IoCompositeProperties prop;
  private static int initialKeyCount = 10;
  private static int mappedKeyCount = 1000;
  private static int uniqueKeyCount = 256;
  private static int mappedValueLength = 20;
  private static int reducedValueLength = 20;
  private static String partitionMethod = "avarage";

  public static final byte mappedElement    = 0x4d; // 'M'
  public static final byte combinedElement  = 0x43; // 'C'
  public static final byte reducedElement   = 0x52; // 'R'


  public static class IoCompositeMapper extends Mapper {

    @Override
    public void configure(Context context) throws Exception {
      Configuration conf = context.getConfiguration();
      initialKeyCount = Integer.parseInt(conf.get("initialKeyCount"));
      mappedKeyCount = Integer.parseInt(conf.get("mappedKeyCount"));
      uniqueKeyCount = Integer.parseInt(conf.get("mappedUniqueKeyCount"));
      mappedValueLength = Integer.parseInt(conf.get("mappedValueLength"));
      reducedValueLength = Integer.parseInt(conf.get("reducedValueLength"));
      partitionMethod = conf.get("partitionMethod");
    }

    private final Pattern p = Pattern.compile("-(\\p{XDigit}+)");
    private void ioCompositeMap(String key, byte[] value,
                                Output<PackableString, PackableBytes> output) throws Exception {

      logger.debug("enter...");
      logger.debug("key = " + key);
      logger.debug("value.length = " + value.length);

      // extract mapperIndex from the filename
      Matcher m = p.matcher(key);
      long mapperIndex = m.find() ? Long.parseLong(m.group(1), 16) : 0L;

      // calculate the number of KVs that this map generates
      long keyCount = (mappedKeyCount + initialKeyCount - mapperIndex - 1) / initialKeyCount;

      // determine the range of uniqueKey for this map
      int outputKeyStart;
      int outputKeyRange;
      if (partitionMethod.equalsIgnoreCase("average")) {
        outputKeyStart = (int)((mappedKeyCount/initialKeyCount)*mapperIndex +
            min(mappedKeyCount % initialKeyCount, mapperIndex));
        outputKeyRange = uniqueKeyCount;
      } else {
        int uKeyCount = (uniqueKeyCount + initialKeyCount - 1) / initialKeyCount
          ;
        int uKeyThres = (uniqueKeyCount + initialKeyCount - 1) % uKeyCount;
        outputKeyStart = 0;
        for (int i = 0; i < mapperIndex; i++) {
          outputKeyStart += (i > uKeyThres) ? (uKeyCount - 1) : uKeyCount;
        }
        outputKeyRange = (mapperIndex > uKeyThres) ? (uKeyCount - 1) : uKeyCount;
      }

      logger.debug(
            " initialKeyCount = " + initialKeyCount + "\n" +
            "  mappedKeyCount = " + mappedKeyCount + "\n" +
            "  uniqueKeyCount = " + uniqueKeyCount + "\n" +
            "--------\n" +
            "     mapperIndex = " + mapperIndex + "\n" +
            "        keyCount = " + keyCount + "\n" +
            "  outputKeyStart = " + outputKeyStart + "\n" +
            "  outputKeyRange = " + outputKeyRange);

      byte[] buf = new byte[mappedValueLength];
      Arrays.fill(buf, mappedElement);
      for (int i = 0; i < keyCount; i++) {
        int ik = (outputKeyStart + i) % outputKeyRange;
        String k = new Integer(ik).toString();
        output.write(new PackableString(k), new PackableBytes(buf));
      }
    }

    public void map(Context context,
                    PackableString key, PackableBytes value,
                    Output<PackableString, PackableBytes> output)
      throws Exception {
      // for debug -->
      mapperCalled();
      // for debug <--
      ioCompositeMap(key.get(), value.get(), output);
    }
  }

  public static class IoCompositeCombiner extends Reducer {
    private PackableBytes result = null;

    @Override
    public void configure(Context context) throws Exception {
      Configuration conf = context.getConfiguration();
      int length = conf.getInt("mappedValueLength");
      byte[] buf = new byte[(int) length];
      Arrays.fill(buf, combinedElement);
      result = new PackableBytes(buf);
    }

    public void reduce(Context context, 
                       PackableString key, Iterable<PackableBytes> values, 
                       Output<PackableString, PackableBytes> output) throws Exception {
      output.write(key, result);
    }
  }

  public static class IoCompositeReducer extends Reducer {

    public void reduce(Context context,
                       PackableString key, Iterable<PackableBytes> values,
                       Output<PackableString, PackableBytes> output) throws Exception {
      logger.debug("enter...");
      logger.debug("reducedValueLength = " + reducedValueLength);
      byte[] buf = new byte[reducedValueLength];
      Arrays.fill(buf, reducedElement);
      output.write(key, new PackableBytes(buf));
    }
  }

  static Class<? extends Mapper> mapperClass = IoCompositeMapper.class;
  static Class<? extends Reducer> combinerClass = IoCompositeCombiner.class;
  static Class<? extends Reducer> reducerClass = IoCompositeReducer.class;
  static Class<? extends Partitioner> partitionerClass = NumPartitioner.class;

  public static GroupID run(SssClient client, GroupID input) throws Exception {
    SimpleJob job = new SimpleJob(client);

// for debug -->
    if (true) {
      Properties pr = prop.getProperties();
      System.err.println("IoCompositeRun#run: ");
      for (Map.Entry<Object, Object> e: pr.entrySet()) {
        String key = (String)e.getKey();
        String value = (String)e.getValue();
        job.getConfiguration().set(key, value);
        System.err.println("  " + key + ": " + value);
      }
    }
// for debug <--

    job.setMapperClass(mapperClass);

    if (prop.getCombinerEnabled()) {
      job.setCombinerClass(combinerClass);
    }

    job.setReducerClass(reducerClass);
    job.setPartitionerClass(partitionerClass);
    job.setInputGID(input);

    return job.execute();
  }

  /**
   * Usage:
   * java java IoCompositeRun groupID
   */
  public static void main(String[] args) throws Exception {
    SssClient client = new SssClient(args);

    String[] others = client.getArgs();
    prop = new IoCompositeProperties(others);

    args = prop.getArgs();
    if (args.length <= 0) {
      System.err.println("Usage: java IoCompositeRun groupID");
      System.exit(1);
    }
    GroupID input = GroupID.createFromString(args[0]);
    long start = System.currentTimeMillis();
    GroupID out = run(client, input);
    long end = System.currentTimeMillis();
    long elapsed = end - start;

    System.out.println("output data deployed - " + out.toString());
    System.out.println("elapsed " + Long.toString(elapsed) + " ms.");

    System.err.println("mapper called = " + mapperCalled + ", reducer called " + reducerCalled);
  }
}

